// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "jobthread.h"

#include <list>
#include <vector>
#include <map>
#include <iostream>

#include "common.h"
#include "getthread.h"
#include "pigasus-types.h"
#include "sendthread.h"
#include "thread.h"
#include "threadpool.h"

using namespace std;

typedef unsigned int uint;

extern ThreadPool *job_thread_pool;
extern pthread_mutex_t *mutex_job_thread_pool;

extern ThreadPool *send_thread_pool;
extern pthread_mutex_t *mutex_send_thread_pool;

extern ThreadPool *get_thread_pool;
extern pthread_mutex_t *mutex_get_thread_pool;

JobThread::JobThread(Servers &servers_, List &common_, 
                     List &inputs_, List *outputs_) {
  type = kJobThreadType;  

  // Make a local copy of the servers. 
  servers = servers_.Clone();
  // Common is expected to be a list of files. These files will be sent to every
  // remote server.
  common = common_.Clone();
  // Inputs is expected to be a list of lists. Each of these sub-lists will be
  // sent to a single remote server.
  inputs = inputs_.Clone();
  // TODO: what was outputs again?
  outputs = outputs_;
}

JobThread::~JobThread() {
  delete servers;
  delete common;
  delete inputs;
  // Do NOT delete outputs---the user needs that.
}

void JobThread::Run() {
  vector<pair<int, Map*> > sent_ids;
  queue<Map *> return_outputs;
  FILE *p;
  char ignore[20];
  char *err;
  string buffer;
  bool done = false;

  if (inputs->GetSize() == 0) {
    inputs->Append(NewList(0));
  }

  // Send out all the files and remotely execute.
  for (int i = 0; i < inputs->GetSize(); i++) {
    Map *current_server = servers->GetNextServer();
    List *inputs_slice = (List*)(inputs->Get(i));
    Thread *send_thread = new SendThread(*current_server, *common,
                                         *inputs_slice);

    // Add SendThread job to SendThreadPool.
    pthread_mutex_lock(mutex_send_thread_pool);
    int sid = send_thread_pool->AddThread(send_thread);
    pthread_mutex_unlock(mutex_send_thread_pool);

    sent_ids.push_back(pair<int, Map*>(sid, current_server));
    debug("JobThread %d: Added SendThread job %d.\n", id, sid); 
  }

  // All jobs have now been sent, we can begin check if any are finished
  // executing remotely, and if so, we can begin to retrieve the files.
  map<int, bool> already_getting;
  for (int i = 0; i < static_cast<int>(sent_ids.size()); i++) {
    already_getting[sent_ids[i].first] = false;
  }

  list<int> get_ids;
  while (true) {
    bool all_done = true;
    for (int i = 0; i < static_cast<int>(sent_ids.size()); i++) {
      if (already_getting[sent_ids[i].first]) {
        continue;
      }

      // Run the checkdone script for this id (sent_ids[i].first).
      
      Map *server = sent_ids[i].second;
      string user = (*(string*)(server->Get("user")));
      string host = (*(string*)(server->Get("host")));
      
      buffer = "./checkdone.sh " + user + " " + host + " " +
               IntToString(sent_ids[i].first);
      p = popen(buffer.c_str(), "r");
      while (!feof(p)) {
        err = fgets(ignore, sizeof(ignore), p);
        done = (ignore[0] == '1') ? true : false;
      }
      pclose(p);
      if (!done) {
        all_done = false;
        continue;
      }
      
      // This sent job slice is done. We can begin downloading the related files
      // from this server and ID.
      Map *out = new Map();
      out->Set("testkey", "testval");
      debug("Map size (at start) is %d\n", out->GetSize());
      debug("Map address is %p\n", out);
      Thread *get_thread = new GetThread(sent_ids[i].first, *(sent_ids[i].second),
                                         out);

      return_outputs.push(out);
      
      // Add get_thread to GetThreadPool.
      pthread_mutex_lock(mutex_get_thread_pool);
      int gid = get_thread_pool->AddThread(get_thread);
      pthread_mutex_unlock(mutex_get_thread_pool);

      get_ids.push_back(gid);
      already_getting[sent_ids[i].first] = true;
    }

    if (all_done) {
      break;
    } else {
      // Sleep for a while before checking again.
      sleep(kDefaultJobThreadDelay);
    }
  }

  // All GetThreads have been sent. Wait for all the files to be downloaded onto
  // the local machine  
  while (!get_ids.empty()) {
    int get_threads_done = 0; 
    for (list<int>::iterator it = get_ids.begin(); it != get_ids.end(); it++) {
      pthread_mutex_lock(mutex_get_thread_pool);
      debug("Gettings status for %d\n", *it);
      int status = get_thread_pool->GetStatus(*it);
      pthread_mutex_unlock(mutex_get_thread_pool);

      if (status == kThreadStatusDone) {
        get_threads_done++;
      }
    }
    
    if (get_threads_done == static_cast<int>(get_ids.size())) {
      break;
    } else {
      sleep(1);
    }
  }
  
  if (outputs != NULL) {
    while (!return_outputs.empty()) {
      debug("Outputs: %s\n",
            ((string*)((return_outputs.front())->Get("stdout")))->c_str());
      outputs->Append(*(return_outputs.front()));
      return_outputs.pop();
    }
  }
  
  pthread_mutex_lock(mutex_job_thread_pool);
  job_thread_pool->SetStatus(id, kThreadStatusMarkAsDone);
  pthread_mutex_unlock(mutex_job_thread_pool);
  debug("Job thread exiting\n");
  return;
}


